package com.open.ssm.mqtt.pub;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.open.ssm.common.utils.spring.SpringUtils;
import com.open.ssm.mqtt.MqttConfig;
import com.open.ssm.mqtt.MqttConstants;

/** 
 * Title:Server 这是发送消息的服务端
 * Description: 服务器向多个客户端推送主题，即不同客户端可向服务器订阅相同主题 
 */  
public class PublishToMqbusMqtt {
	
	protected static Logger logger = LoggerFactory.getLogger(PublishToMqbusMqtt.class);
	
	private int QOS = 1; // 定义MQTT的qos
	private static String maxPoll = "10"; // 最大连接池大小
	private static List<MqttClient> clientPool = new ArrayList<MqttClient>();
	// 单例模式
	private static PublishToMqbusMqtt pubInstance = null;
	// mqtt的参数配置
	private static MqttConfig mqttConfig = SpringUtils.getBean(MqttConfig.class);
	   
	private PublishToMqbusMqtt() {
	}
	
	public static PublishToMqbusMqtt getInstance() {
	   if (pubInstance == null) {
		   pubInstance = new PublishToMqbusMqtt();
	   }
	   //初始化mqtt连接池pool
	   if (clientPool == null || clientPool.size()==0) {
		   inintPool();
	   }
	   
	   return pubInstance;
	}

	 /** 
         * 初始化发送链接池子 
     */  
    private static void inintPool() {
    	String serverURI = mqttConfig.getServerURI(); 
    	String userName =  mqttConfig.getUsername();
        String password =  mqttConfig.getPassword();
        int connectionTimeout = mqttConfig.getConnectionTimeout();
        int keepAliveInterval = mqttConfig.getKeepAliveInterval();
	   //清空clientPool.
	   if(clientPool!=null) {
		   clientPool.clear();
	   }
	  
	   for(int i=1;i<=Integer.parseInt(maxPoll);i++) {
		   MqttClient client=null;
		   try {
				client = new MqttClient(serverURI, MqttConstants.PUB_DEVICE_CLIENTID+"_"+i, new MemoryPersistence());
			} catch (MqttException e) {
				e.printStackTrace();
			}  
		    //1：设置option参数
	        MqttConnectOptions options = new MqttConnectOptions();  
	        options.setUserName(userName);  
	        options.setPassword(password.toCharArray());  
	        // 设置超时时间  
	        options.setConnectionTimeout(connectionTimeout);  
	        // 设置会话心跳时间  
	        options.setKeepAliveInterval(keepAliveInterval); 
	        //清除会话
	        options.setCleanSession(true);
	        try {  
	            client.setCallback(new PushMqCallback());  
	            client.connect(options);  
	        } catch (Exception e) {  
	        	logger.error(e.getMessage(),e);
	        }  
	        clientPool.add(client);
	   }
    }
    /** 
     * 发布者发送消息 
     * @param topic 
     * @param message 
     */  
    public boolean publishMsg(String topic, String sendMsg) {
    	byte[] message = null;
		try {
			message = sendMsg.getBytes("UTF-8");
		} catch (UnsupportedEncodingException e1) {
			logger.error(e1.getMessage(),e1);
		}
    	boolean ret=false;
    	
    	//组装消息
    	MqttMessage mqtMsg = new MqttMessage();  
    	mqtMsg.setQos(QOS);  //保证消息能到达一次
    	mqtMsg.setPayload(message);   
    	
    	//获得链接池中随机的一个链接对象
    	Random random = new Random();
    	int sindex = random.nextInt(Integer.parseInt(maxPoll));
    	//初始化mqtt连接池pool
 	    if (clientPool == null || clientPool.size()==0) {
 		   inintPool();
 	    }
 	    if(sindex > (clientPool.size()-1)){//防止多线程溢出index
 	    	sindex=0;
 	    }
    	MqttClient clientRandom=clientPool.get(sindex);
 	   //clientRandom为空 则表示，有其他服务器节点已经启动占用了链接资源，需要重新 初始化mqtt连接池pool
 	   if (clientRandom == null || clientRandom.isConnected()==false) {
 		  //初始化mqtt连接池pool
 		  inintPool();
 		  clientRandom=clientPool.get(sindex); 
 	   }
    	//发送消息
    	try {
    		clientRandom.publish(topic , mqtMsg);
    		logger.info("PublishToMqbusMqtt发布消息topic:{},content:{}", topic, sendMsg);
		} catch (MqttException e) {
			logger.error(e.getMessage(),e);
			return ret;
		}
    	ret=true;
    	return ret;
    }  

//	  /** 
//     *  启动入口 
//     * @param args 
//     * @throws MqttException 
//     */  
//    public static void main(String[] args) throws MqttException { 
//    	
//    	//测试发布者 消息
//        //持续发送测试
////        for(int i=0;i<5;i++){
////      	  
////            //PB序列化使用
////            //1：序列化数据部对象
////            System.out.println("===== 构建一个PB模型开始 =====");
////            
////            
////            //1:充电桩 离线交易记录上传test
////            
////            //2:赋值数据部对象给APDU
////            ApduOuterClass.Apdu.Builder apdu = ApduOuterClass.Apdu.newBuilder();
////            apdu.setCmdType(CmdType.TRADERECORDS);//业务类型标志
////            apdu.setEquipNo("1000101170068101");//充电设备编号
////            apdu.setConnectId(0);//充电接口标识
////            apdu.setType(0);//设备类型
////            apdu.setTimeStamp(new Date().getTime());//时间戳
////            apdu.setSnno(0);//序列号
////            apdu.setEnc(EncType.NOENC);//数据域加密方式 
//////            apdu.setDataArea(dataArea_data.toByteString());//数据域字节
////            
////            ApduOuterClass.Apdu apdu_data = apdu.build();
////            System.out.println( apdu_data.toString());
////            System.out.println("===== 构建一个PB模型结束 =====");
////            System.out.println("\n" + "bytes长度" +  apdu_data.toByteString().size());
////            
////            //发送
////            String TOPIC = "mqtt/command/0000101170068101"; 
////            
//////            TOPIC = "chgequ/s2c/0000101170068101"; 
////            
////            PublishToEquipMqtt.getInstance().publishMsg("0000101170068101",TOPIC,apdu_data.toByteArray());
////            
////            //间隔5s
////            try {
////				Thread.sleep(1000*5);
////			  } catch (InterruptedException e) {
////			  }
////        }
//    	
//    }
    
}  

